/* 
 * Copyright [2018] [Alex/libo(liboware@gmail.com)]
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *  http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hyts.reactor.produce;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.hyts.reactor.base.ProduceManager;
import com.hyts.reactor.constant.DisruptorConfiguration;
import com.hyts.reactor.model.EventModel;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.LiteBlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import lombok.Getter;
import lombok.Setter;
import lombok.experimental.Accessors;

/** 
 * @author LiBo/Alex
 * @version V1.0 
 */
//@RequiredArgsConstructor
@Accessors(chain=true)
public class DisruptorProduceManager<T extends EventFactory> extends ProduceManager<T>{
	
	@Getter
	@Setter
	private ExecutorService executorService;

	@Getter
	@Setter
	private WaitStrategy definedWaitStrategy;
	
	@Getter
	@Setter
	private ProducerType productType;
	
	@Getter
	private final EventModel<T> model;
	
	@Setter	@Getter
	private Disruptor<T> disruptor;

	@Setter
	private int ringbufferSize;
	
	/**  
	 */
	@SuppressWarnings("unchecked")
	public DisruptorProduceManager(T object) {
		super();
		this.executorService = Executors.newCachedThreadPool();
		this.definedWaitStrategy = new LiteBlockingWaitStrategy();
		this.productType = ProducerType.SINGLE;
		this.model = new EventModel<T>(object);
		disruptor = new Disruptor<T>(model.getData(),ringbufferSize < 1 ? 
				DisruptorConfiguration.DEFAULT_RINGBUFFER_SIZE:
					ringbufferSize,executorService,productType,definedWaitStrategy);
	}
	
	
	@SuppressWarnings("unchecked")
	public DisruptorProduceManager(T object,EventHandler<T> handler) {
		super();
		this.executorService = Executors.newCachedThreadPool();
		this.definedWaitStrategy = new LiteBlockingWaitStrategy();
		this.productType = ProducerType.SINGLE;
		this.model = new EventModel<T>(object);
		disruptor = new Disruptor<T>(model.getData(),ringbufferSize < 1 ? 
				DisruptorConfiguration.DEFAULT_RINGBUFFER_SIZE:
					ringbufferSize,executorService,productType,definedWaitStrategy);
		disruptor.handleEventsWith(handler);
	}
	
	@SuppressWarnings("unchecked")
	public DisruptorProduceManager(T object,EventHandler<T>... handler) {
		super();
		this.executorService = Executors.newCachedThreadPool();
		this.definedWaitStrategy = new LiteBlockingWaitStrategy();
		this.productType = ProducerType.SINGLE;
		this.model = new EventModel<T>(object);
		disruptor = new Disruptor<T>(model.getData(),ringbufferSize < 1 ? 
				DisruptorConfiguration.DEFAULT_RINGBUFFER_SIZE:
					ringbufferSize,executorService,productType,definedWaitStrategy);
		disruptor.handleEventsWith(handler);
	}
	
	
	@SuppressWarnings("unchecked")
    public DisruptorProduceManager(T object,WorkHandler<T>... handler) {
        super();
        this.executorService = Executors.newCachedThreadPool();
        this.definedWaitStrategy = new LiteBlockingWaitStrategy();
        this.productType = ProducerType.SINGLE;
        this.model = new EventModel<T>(object);
        disruptor = new Disruptor<T>(model.getData(),ringbufferSize < 1 ? 
                DisruptorConfiguration.DEFAULT_RINGBUFFER_SIZE:
                    ringbufferSize,executorService,productType,definedWaitStrategy);
        disruptor.handleEventsWithWorkerPool(handler);
    }
    
	/**  
	 * @return
	 * @exception
	 */ 
	public RingBuffer<T> start() {
		//@SuppressWarnings("unused")
		/*DisruptorProduceManager<T> manager = 
				new DisruptorProduceManager<T>(model.getData());*/
		return disruptor.start();
	}
	
	/**  
	 * @return
	 * @exception
	 */ 
	public RingBuffer<T> getElement(){
		return disruptor.getRingBuffer();
	}
	
	/**
	 * @param exceptionHandler
	 * @return
	 * @exception
	 */
	public DisruptorProduceManager<?> addExceptionHandler(ExceptionHandler exceptionHandler) {
	    disruptor.handleExceptionsWith(exceptionHandler);
	    return this;
	}
	

	public static class User implements EventFactory{
		
		public String name;
		
		/* (�� Javadoc)
		 * @return
		 * @see com.lmax.disruptor.EventFactory#newInstance()  
		 * @exception
		 */ 
		
		
		@Override
		public User newInstance() {
			return new User();
		}

		public String getName() {
			return name;
		}

		public void setName(String name) {
			this.name = name;
		}
	}
	
	
	public static void main(String[] args) {
		DisruptorProduceManager<User> manager = 
				new DisruptorProduceManager<User>(new User(),(user,param1,param2)->{
			System.out.println(user.getName());
		});
		//manager.getDisruptor().handleEventsWith();
		manager.start();
		RingBuffer<User> ringBuffer = manager.getElement();
		for (long l = 0; true; l++) {
			String star = String.valueOf(l);
			ringBuffer.publishEvent((event, sequence) -> event.setName("2342342342"+star));
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
	
}
